package com.gitee.ixtf.rsocket.kotlinx

import cn.hutool.core.util.ReflectUtil
import com.gitee.ixtf.core.kotlinx.CLOUD_EVENT_FORMAT
import com.gitee.ixtf.core.kotlinx.readJson
import com.gitee.ixtf.core.kotlinx.serialize
import io.cloudevents.CloudEvent
import io.cloudevents.core.builder.CloudEventBuilder
import io.netty.buffer.ByteBufAllocator
import io.netty.buffer.ByteBufUtil
import io.netty.buffer.Unpooled
import io.netty.util.ReferenceCountUtil
import io.rsocket.DuplexConnection
import io.rsocket.Payload
import io.rsocket.RSocket
import io.rsocket.metadata.RoutingMetadata
import io.rsocket.metadata.TaggingMetadataCodec
import io.rsocket.util.DefaultPayload
import io.vertx.core.buffer.Buffer
import io.vertx.core.json.JsonArray
import io.vertx.core.json.JsonObject
import java.net.SocketAddress
import kotlinx.coroutines.reactor.awaitSingle
import kotlinx.coroutines.reactor.mono

fun CloudEventBuilder.payload(route: String): Payload = build().payload(route)

fun CloudEvent.payload(route: String): Payload =
    TaggingMetadataCodec.createTaggingContent(ByteBufAllocator.DEFAULT, listOf(route)).let {
      DefaultPayload.create(Unpooled.wrappedBuffer(serialize()), it)
    }

fun Payload.route(): String = RoutingMetadata(metadata()).first()

fun CloudEventBuilder.withData(data: Buffer) = apply { withData(data.bytes) }

fun CloudEventBuilder.withData(data: JsonObject) = apply { withData(data.toBuffer()) }

fun CloudEventBuilder.withData(data: JsonArray) = apply { withData(data.toBuffer()) }

suspend fun Payload.cloudEvent(): CloudEvent = CLOUD_EVENT_FORMAT.deserialize(bytes())

suspend inline fun <reified T> Payload.readJson(): T = bytes().readJson()

suspend fun Payload.bytes(): ByteArray =
    mono { ByteBufUtil.getBytes(sliceData()) }
        .doFinally { ReferenceCountUtil.safeRelease(this) }
        .awaitSingle()

fun ByteArray.payload(): Payload = DefaultPayload.create(this)

suspend fun Payload.buffer(): Buffer = Buffer.buffer(bytes())

fun Buffer.payload() = bytes.payload()

suspend fun Payload.jsonObject() = JsonObject(buffer())

fun JsonObject.payload(): Payload = toBuffer().payload()

suspend fun Payload.jsonArray() = JsonArray(buffer())

fun JsonArray.payload(): Payload = toBuffer().payload()

fun RSocket.remoteAddress(): SocketAddress? =
    runCatching { ReflectUtil.getFieldValue(this, "connection") }
        .map { (it as? DuplexConnection)?.remoteAddress() }
        .getOrElse { null }
